package com.leilei.support.redis;

import com.leilei.support.util.CollectionSplitUtil;
import net.oschina.j2cache.Level2Cache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;

import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author lei
 * @create 2022-05-11 10:04
 * @desc 当l2 redis缓存设置为generic时当前类有效，支持String数据结构操作，且可以为每个单独的key设置过期时间（与一级缓存保持一致）
 **/
public class SpringRedisGenericCache implements Level2Cache {

    private final static Logger log = LoggerFactory.getLogger(SpringRedisGenericCache.class);

    private final String namespace;

    private final String region;

    private final RedisTemplate<String, Serializable> redisTemplate;

    public SpringRedisGenericCache(String namespace, String region, RedisTemplate<String, Serializable> redisTemplate) {
        // 缺省region
        if (region == null || region.isEmpty()) {
            region = "_";
        }
        this.namespace = namespace;
        this.redisTemplate = redisTemplate;
        this.region = getRegionName(region);
    }

    /**
     * 在region里增加一个可选的层级,作为命名空间,使结构更加清晰
     * 同时满足小型应用,多个J2Cache共享一个redis database的场景，这个时候就可以根据项目名定义name-space，进行缓存区分
     *
     * @param region
     * @return
     */
    private String getRegionName(String region) {
        if (namespace != null && !namespace.isEmpty()) {
            region = namespace + ":" + region;
        }
        return region;
    }

    /**
     * 清理整个region时触发
     */
    @Override
    public void clear() {
        Collection<String> keys = keys();
        if (keys.size() > 1) {
            Collection<Set<String>> sets = CollectionSplitUtil.split(keys);
            sets.forEach(set -> redisTemplate.delete(set.stream().map(x -> this.region + ":" + x).collect(Collectors.toSet())));
        } else {
            keys.forEach(k -> redisTemplate.delete(this.region + ":" + k));
        }
    }

    @Override
    public boolean exists(String key) {
        Boolean execute = redisTemplate.execute((RedisCallback<Boolean>) redis -> redis.exists(buildKey(key)));
        return execute != null && execute;
    }

    /**
     * 清除指定缓存KEY
     * @param keys
     */
    @Override
    public void evict(String... keys) {
        byte[][] bytes = Stream.of(keys).map(this::buildKey).toArray(byte[][]::new);
        redisTemplate.execute((RedisCallback<Long>) redis -> redis.del(bytes));
        // for (String k : keys) {
        //     redisTemplate.execute((RedisCallback<Long>) redis -> redis.del(buildKey(k)));
        // }
    }

    /**
     * 获取 region中所有缓存KEY,取消keys 更改为scan
     *
     * @return
     */
    @Override
    public Collection<String> keys() {
        Set<String> result = redisTemplate.execute((RedisCallback<Set<String>>) cc -> {
            Set<String> keys = new HashSet<>();
            // count 并不是只获取指定数量个，而是告诉redis一次扫描多少个字典槽
            ScanOptions build = ScanOptions.scanOptions().count(1024).match(this.region + ":*").build();
            // 我们之只需要判断返回的游标是否有值即可判断是否还有余下元素
            Cursor<byte[]> scan = cc.scan(build);
            while (scan.hasNext()) {
                keys.add(new String(scan.next()));
            }
            return keys;
        });
        if (result == null) {
            return new HashSet<>();
        }
        return result.stream().map(k -> k.substring(this.region.length() + 1)).collect(Collectors.toSet());
    }

    @Override
    public byte[] getBytes(String key) {
        return redisTemplate.execute((RedisCallback<byte[]>) redis -> redis.get(buildKey(key)));
    }

    @Override
    public List<byte[]> getBytes(Collection<String> keys) {
        return redisTemplate.execute((RedisCallback<List<byte[]>>) redis -> {
            byte[][] bytes = keys.stream().map(this::buildKey).toArray(byte[][]::new);
            return redis.mGet(bytes);
        });
    }

    @Override
    public void setBytes(String key, byte[] bytes, long timeToLiveInSeconds) {
        if (timeToLiveInSeconds <= 0) {
            log.debug(String.format("Invalid timeToLiveInSeconds value : %d , skipped it.", timeToLiveInSeconds));
            setBytes(key, bytes);
        } else {
            redisTemplate.execute((RedisCallback<List<byte[]>>) redis -> {
                redis.setEx(buildKey(key), (int) timeToLiveInSeconds, bytes);
                return null;
            });
        }
    }

    @Override
    public void setBytes(Map<String, byte[]> bytes, long timeToLiveInSeconds) {
        bytes.forEach((k, v) -> setBytes(k, v, timeToLiveInSeconds));
    }

    @Override
    public void setBytes(String key, byte[] bytes) {
        redisTemplate.execute((RedisCallback<byte[]>) redis -> {
            redis.set(buildKey(key), bytes);
            return null;
        });
    }

    @Override
    public void setBytes(Map<String, byte[]> bytes) {
        bytes.forEach(this::setBytes);
    }

    private byte[] buildKey(String key) {
        byte[] k;
        k = (this.region + ":" + key).getBytes(StandardCharsets.UTF_8);
        return k;
    }
}
